﻿unit ContextFactory;

interface

uses
  Winapi.Windows, Engine, WinSocket, WinSock2Ex, CriticalSection, Protocol,
  System.SysUtils, Winapi.ActiveX, MemBuffer, GlobalMonitor, EventHandler;

type
  TIMapClientContext = class;
  TIClientContext = class;
  TIOCPDecoder = class(TObject)
  public
    /// <summary>
    ///   解码收到的数据,如果有接收到数据,调用该方法,进行解码
    /// </summary>
    /// <returns>
    ///   返回解码好的对象
    /// </returns>
    /// <param name="inBuf"> 接收到的流数据 </param>
    function Decode(const inBuf: TIBuffer): TObject; virtual; abstract;
  end;

  TIOCPEncoder = class(TObject)
  public
    /// <summary>
    ///   编码要发生的对象
    /// </summary>
    /// <param name="pvDataObject"> 要进行编码的对象 </param>
    /// <param name="ouBuf"> 编码好的数据 </param>
    procedure Encode(pvDataObject:TObject; const ouBuf: TIBuffer); virtual; abstract;
  end;
  TIClientContext = class(TIEventHandler)
  private
    ConnTime: Cardinal;
    PostCloseIO: Boolean;
    MapClientContext: TIMapClientContext;
    FSocketConn: Boolean;
    FConnDateTime: TDateTime;
    FSessionID: TGUID;
    FMapID: Cardinal;
    FContextLock: TCriticalSection;
    procedure SetMapID(const Value: Cardinal);
    procedure SetSessionID(const Value: TGUID);
    //设置连接状态
    procedure SetClientConnState(value: Boolean = True);
    //投递关闭Socket信息, 等待所有未决的I/O返回
    function PostWSACloseSocket: Boolean;
    //投递SEND I/O消息, 发送消息
    function PostWSASendEvent(FSendBuffer: TIBuffer): Boolean;
    //投递RECV I/O消息, 开始接收用户数据
    function PostWSARecvSocket: Boolean;
    //投递Recv I/O 消息
    function PostWSARecv(FRecvBuffer: TIBuffer; m_ProtType: TSERVER_PROTOOOL_TYPE = PROTOOOL_TCP): IO_RECV_STATUS;
    //投递Send I/O 消息
    function PostWSASend(FSendBuffer: TIBuffer; m_ProtType: TSERVER_PROTOOOL_TYPE = PROTOOOL_TCP): IO_SEND_STATUS;
  protected
    Engine: TIEngine;
  public
    FSocket: TSocket;
    RemoteIP: AnsiString;
    RemotePort: Integer;
    NextContext: TIClientContext;
    //设置套接字及GUID
    procedure SetSocket(ASocket: TSocket);
    //套接字连接消息
    procedure SocketConnect;
    //套接字断开消息
    procedure CloseSocket;
    //处理RECV I/O消息
    function OnWSARecv(FRecvBuffer: TIBuffer; dwIoSize: Cardinal): Boolean; virtual;
    //处理SEND I/O消息
    function OnWSASend(FSendBuffer: TIBuffer; dwIoSize: Cardinal): Boolean; virtual;
    //回池上下文到上下文列表
    procedure ReleaseToPool;
    //从内存池分配一个TIBuffer
    function GetFreeDIBuffer(IOType: IO_POST_EVENT_TYPE): TIBuffer;
    //发送数据流
    procedure SendStream(pData: Pointer; nLenth: Integer); overload;
    procedure SendStream(FSendBuffer: TIBuffer); overload;
    function Connect(strIPAddr: AnsiString; nPort: Cardinal): Boolean;
    //数据处理
    procedure ProcessDataStream(FBuffer: TIBuffer); virtual;
    //初始化Session
    procedure InitClientContext; virtual;
    constructor Create(FEngine: TIEngine); overload; virtual;
    destructor Destroy; override;
  published
    property MapID: Cardinal read FMapID write SetMapID;
    property SessionID: TGUID read FSessionID write SetSessionID;
    property SocketConn: Boolean read FSocketConn write FSocketConn;
    property ConnDateTime: TDateTime read FConnDateTime write FConnDateTime;
    property ContextLock: TCriticalSection read FContextLock;
  end;
  //创建上下文事件
  TOnCreateClientEvent = function: TIClientContext of object;
  (* TDIClientContext节点   *)
  PTDIClientContextNode  =  ^TDIClientContextNode;
  TDIClientContextNode   =  record
    MapID:               Cardinal;                 (* MapID            *)
    IsUsed:              Boolean;               (* 是否为有效节点   *)
    IsRelease:           Boolean;               (* 释放             *)
    ClientContext:       TIClientContext;      (* TIClientContext *)
  end;
  TDIClientContextArray  =  array of TDIClientContextNode;
  //用户管理
  TIMapClientContext = class
  private
    FClientContextCount:  WORD;                  (* 数量             *)
    FMaxCount:            WORD;                  (* 实际最大索引     *)
    FActiveCount:         WORD;                  (* 在线数量         *)
    FBuckets:              TDIClientContextArray; (* 上下文数组       *)
    FMapClientContextLock: TCriticalSection;      (* 锁               *)
    FOnCreateClientEvent:  TOnCreateClientEvent;  (* 创建上下文事件   *)
  published
    property ClientContextCount: WORD read FClientContextCount;
    property ActiveClientCount: WORD read FActiveCount;
    property MaxCount: WORD read FMaxCount;
    property Buckets: TDIClientContextArray read FBuckets;
    property MapClientContextLock: TCriticalSection read FMapClientContextLock;
    property OnCreateClientEvent: TOnCreateClientEvent read FOnCreateClientEvent
      write FOnCreateClientEvent;
  public
    procedure InitHasTableLength(Count: WORD);
    function AllocateFreeClientContextFromPool: TIClientContext;
    function RemoveClientContext(FClientContext: TIClientContext): Boolean;
    function FindClientContext(Index: Cardinal): TIClientContext;
    procedure FreeClientContexts;
  public
    constructor Create;
    destructor Destroy; override;
  end;

implementation

{ TIClientContext }

procedure TIClientContext.CloseSocket;
var
  lingerStrt: TLinger;
begin
  FContextLock.Lock;
  //关闭套接字
  if FSocket <> INVALID_SOCKET then
  begin
    //设置强制关闭Socket连接,否则会等待Socket发送数据完毕或超时
    lingerStrt.l_onoff := 1;
    lingerStrt.l_linger := 0;
    WinSock2Ex.setsockopt(FSocket, SOL_SOCKET, SO_LINGER, @lingerStrt, SizeOf(lingerStrt));
    //CloseSocket套接字, 取消完成端口中的I/O
    CancelIO(FSocket);
    WinSocket.CloseWinSocket(FSocket);
    SetClientConnState(False);
    PostCloseIO := False;
    with Engine do
      if Assigned(OnChannelCloseEvent) then OnChannelCloseEvent(Self);
  end;
  PostWSACloseSocket;
  FContextLock.UnLock;
end;

function TIClientContext.Connect(strIPAddr: AnsiString; nPort: Cardinal): Boolean;
var
  bVal: Boolean;
  ASocket: TSocket;
begin
  Result := False;
  if not CreateWSASocket(ASocket) then Exit;
  if WinSocket.Connect(ASocket, strIPAddr, nPort) then
  begin
    //检查套接字
    if (ASocket = INVALID_SOCKET) then Exit;
    //设置Socket, KeyID, 得到客户端IP地址, 设置GUID
    SetSocket(ASocket);
    (* 禁止nagle算法 *)
    bVal := True;
    if not SetSocketNagle(ASocket, bVal) then
    begin
      CloseWinSocket(ASocket);
      Exit;
    end;
    //关联完成端口
    if Engine.IOCompletionPort.AssociateSocketWithCompletionPort(ASocket, Cardinal(Pointer(Self))) then
      begin
        PostWSARecvSocket;  //投递RECV I/O接受客户端数据
        Result := True;
      end
    else
      CloseWinSocket(ASocket);
  end;
end;

constructor TIClientContext.Create(FEngine: TIEngine);
begin
  inherited Create;
  Engine := FEngine;
  FContextLock := TCriticalSection.Create;
  FContextLock.SetLockName('ContextLock');
  FSocket := INVALID_SOCKET;
  NextContext := nil;
  SocketConn := False;
  PostCloseIO := False;
  ConnTime := 0;
end;

destructor TIClientContext.Destroy;
begin
  FreeAndNil(FContextLock);
  inherited Destroy;
end;

function TIClientContext.GetFreeDIBuffer(IOType: IO_POST_EVENT_TYPE): TIBuffer;
begin
  Result := Engine.MemoryPool.AllocateBuffer(IOType);
end;

procedure TIClientContext.InitClientContext;
begin
  FSocket := INVALID_SOCKET;
  NextContext := nil;
  SocketConn := False;
  PostCloseIO := False;
  ConnTime := 0;
end;

function TIClientContext.OnWSARecv(FRecvBuffer: TIBuffer; dwIoSize: Cardinal): Boolean;
begin
  Result := True;
  FRecvBuffer.SetUsed(dwIoSize);
  with Engine do
    if Assigned(OnChannelRecvEvent) then
      OnChannelRecvEvent(Self, FRecvBuffer, dwIoSize);
  //回收RECV Buffer
  PostWSARecvSocket;
  _GlobalMonitor.PortIOCPOnRecv(dwIoSize);
  Engine.MemoryPool.ReleaseBuffer(FRecvBuffer);
end;

function TIClientContext.OnWSASend(FSendBuffer: TIBuffer; dwIoSize: Cardinal): Boolean;
begin
  Result := True;
  FSendBuffer.SetUsed(dwIoSize);
  with Engine do
    if Assigned(OnChannelSendEvent) then
      OnChannelSendEvent(Self, FSendBuffer, dwIoSize);
  //回收SEND Buffer
  _GlobalMonitor.PortIOCPOnSend(dwIoSize);
  Engine.MemoryPool.ReleaseBuffer(FSendBuffer);
end;

function TIClientContext.PostWSACloseSocket: Boolean;
var
  CloseBuffer: TIBuffer;
begin
  Result := False;
  if not PostCloseIO then
  begin
    PostCloseIO := True;
    CloseBuffer := GetFreeDIBuffer(IO_WSA_CLOSESOCKET);
    CloseBuffer.SetupCloseRead;
    CloseBuffer.OverLappedEx.FGUID := FSessionID;
    if not Engine.IOCompletionPort.PostIOCompletionStatus(1, Cardinal(Pointer(Self)),
      @CloseBuffer.OverLappedEx) then
    begin
      PostCloseIO := False;
      Exit;
    end;
    Result := True;
  end;
end;

function TIClientContext.PostWSARecv(FRecvBuffer: TIBuffer;
  m_ProtType: TSERVER_PROTOOOL_TYPE): IO_RECV_STATUS;
var
  nRet, nError: Integer;
  dwFlags, dwIOSize: Cardinal;
begin
  dwFlags := 0;
  dwIOSize := 0;
  nRet := -1;
  case m_ProtType of
  PROTOOOL_TCP:
    nRet := WinSock2Ex.WSARecv(FSocket, @FRecvBuffer.FWSABuf, 1, dwIOSize,
      dwFlags, @FRecvBuffer.OverLappedEx, nil);
  PROTOOOL_UDP:
    nRet := WinSock2Ex.WSARecvFrom(FSocket, @FRecvBuffer.FWSABuf, 1, @dwIOSize,
      dwFlags, @FRecvBuffer.UDPAddr, @FRecvBuffer.UDPAddrLen, @FRecvBuffer.OverLappedEx,
       nil);
  end;
  if (nRet = SOCKET_ERROR) then
    begin
      nError := WSAGetLastError();
      case nError of
      WSA_IO_PENDING: Result := IO_RECV_IO_PENDING;
      WSAEWOULDBLOCK: Result := IO_RECV_WOULDBLOCK;
      WSAENETRESET:   Result := IO_RECV_RESET;
      WSAECONNRESET:  Result := IO_RECV_RESET;
      else
        Result := IO_RECV_ERROR;
      end;
    end
  else
    Result := IO_RECV_SUCCESS;
end;

function TIClientContext.PostWSARecvSocket: Boolean;
var
  FRecvBuffer: TIBuffer;
  dwIOStatus: IO_RECV_STATUS;
begin
  Result := True;
  with Engine do
  begin
    if (FSocket <> INVALID_SOCKET) and (not PostCloseIO) and Started and (not ShutDown) then
    begin
      //设置客户端Session
      FRecvBuffer := GetFreeDIBuffer(IO_WSA_RECV);
      FRecvBuffer.OverLappedEx.FGUID := FSessionID;
      FRecvBuffer.SetupRead;
      _GlobalMonitor.ProcIOCPPostRecv;
      dwIOStatus := PostWSARecv(FRecvBuffer);
      if ((dwIOStatus <> IO_RECV_SUCCESS) and (dwIOStatus <> IO_RECV_IO_PENDING)) then
      begin
        CloseSocket;
        MemoryPool.ReleaseBuffer(FRecvBuffer);
        Result := False;
      end;
    end;
  end;
end;

function TIClientContext.PostWSASend(FSendBuffer: TIBuffer;
  m_ProtType: TSERVER_PROTOOOL_TYPE): IO_SEND_STATUS;
var
  nRet, nError: Integer;
  dwFlags, dwIOSize: Cardinal;
begin
  dwIOSize := 0;
  dwFlags := 0;
  nRet := -1;
  case m_ProtType of
  PROTOOOL_TCP:
    nRet := WinSock2Ex.WSASend(FSocket, @FSendBuffer.FWSABuf, 1, dwIOSize,
      dwFlags, @FSendBuffer.OverLappedEx, nil);
  PROTOOOL_UDP:
    nRet := WinSock2Ex.WSASendTo(FSocket, @FSendBuffer.FWSABuf, 1, @dwIOSize,
      dwFlags, FSendBuffer.UDPAddr, SizeOf(TSockAddr), @FSendBuffer.OverLappedEx,
       nil);
  end;
  if (nRet = SOCKET_ERROR) then
    begin
      nError := WSAGetLastError();
      case nError of
      WSA_IO_PENDING: Result := IO_SEND_IO_PENDING;
      WSAEWOULDBLOCK: Result := IO_SEND_WOULDBLOCK;
      WSAENETRESET:   Result := IO_SEND_RESET;
      WSAECONNRESET:  Result := IO_SEND_RESET;
      else
        Result := IO_SEND_ERROR;
      end;
    end
  else
    Result := IO_SEND_SUCCESS;
end;

function TIClientContext.PostWSASendEvent(FSendBuffer: TIBuffer): Boolean;
var
  dwIOStatus: IO_SEND_STATUS;
begin
  Result := True;
  with Engine do
  begin
    if (FSocket <> INVALID_SOCKET) and (not PostCloseIO) and Engine.Started and (not Engine.ShutDown) then
    begin
      FSendBuffer.OverLappedEx.FGUID := FSessionID;
      FSendBuffer.SetOperation(IO_WSA_SEND);
      FSendBuffer.SetupWrite;
      _GlobalMonitor.ProcIOCPPostSend;
      dwIOStatus := PostWSASend(FSendBuffer);
      if ((dwIOStatus <> IO_SEND_SUCCESS) and (dwIOStatus <> IO_SEND_IO_PENDING)) then
      begin
        CloseSocket;
        MemoryPool.ReleaseBuffer(FSendBuffer);
        Result := False;
      end;
    end;
  end;
end;

procedure TIClientContext.ProcessDataStream(FBuffer: TIBuffer);
begin
  ;
end;

procedure TIClientContext.ReleaseToPool;
begin
  MapClientContext.RemoveClientContext(Self);
end;

procedure TIClientContext.SendStream(FSendBuffer: TIBuffer);
begin
  PostWSASendEvent(FSendBuffer);
end;

procedure TIClientContext.SendStream(pData: Pointer; nLenth: Integer);
var
  TemData: Pointer;
  SequenceNumber, SendLen: WORD;
  NeedSendLen: Integer;
  FSendBuffer: TIBuffer;
begin
  //m_SendDataLock.Lock;
  try
    if nLenth <= MAX_PACKAGESIZE then
      begin
        FSendBuffer := GetFreeDIBuffer(IO_WSA_SEND);
        FSendBuffer.CreatePackage(pData, nLenth);
        PostWSASendEvent(FSendBuffer);
      end
    else
      begin
        TemData := pData;
        NeedSendLen := nLenth;
        FSendBuffer := GetFreeDIBuffer(IO_WSA_SEND);
        FSendBuffer.CreatePackage(TemData, MAX_PACKAGESIZE);
        PostWSASendEvent(FSendBuffer);
        NeedSendLen := NeedSendLen - MAX_PACKAGESIZE;
        TemData := Pointer(Cardinal(TemData) + MAX_PACKAGESIZE);
        SequenceNumber := 1;
        while NeedSendLen > 0 do
        begin
          SendLen := MAX_PACKAGESIZE;
          if SendLen >= NeedSendLen then SendLen := NeedSendLen;
          FSendBuffer := GetFreeDIBuffer(IO_WSA_SEND);
          FSendBuffer.SetSequenceNumber(SequenceNumber);
          FSendBuffer.CreatePackage(TemData, SendLen);
          PostWSASendEvent(FSendBuffer);
          NeedSendLen := NeedSendLen - SendLen;
          TemData := Pointer(Cardinal(TemData) + SendLen);
          inc(SequenceNumber);
        end;
      end;
  finally
    //m_SendDataLock.UnLock;
  end;
end;

procedure TIClientContext.SetClientConnState(value: Boolean);
begin
  if value then
    begin
      SocketConn := True;
      ConnDateTime := Now();
      ConnTime := GetTickCount;
    end
  else
    SocketConn := False;
end;

procedure TIClientContext.SetMapID(const Value: Cardinal);
begin
  FMapID := Value;
end;

procedure TIClientContext.SetSessionID(const Value: TGUID);
begin
  FSessionID := Value;
end;

procedure TIClientContext.SetSocket(ASocket: TSocket);
begin
  FSocket := ASocket;
  CoCreateGUID(FSessionID);
end;

procedure TIClientContext.SocketConnect;
begin
  SetClientConnState(True);
  with Engine do
    if Assigned(OnChannelOpenEvent) then OnChannelOpenEvent(Self);
  PostWSARecvSocket;
end;

{ TIMapClientContext }

function TIMapClientContext.AllocateFreeClientContextFromPool: TIClientContext;
var
  i: Integer;
begin
  Result := nil;
  FMapClientContextLock.Lock;
  for i := 1 to FClientContextCount do
  begin
    if (not FBuckets[i].IsUsed) then
    begin
      FBuckets[i].IsUsed := True;
      Result := FBuckets[i].ClientContext;
      Result.FMapID := FBuckets[i].MapID;
      Result.InitClientContext;
      Inc(FActiveCount);
      if FMaxCount < i then FMaxCount := i;
      Break;
    end;
  end;
  FMapClientContextLock.UnLock;
end;

constructor TIMapClientContext.Create;
begin
  inherited Create;
  FActiveCount := 0;
  FMaxCount := 0;
  FMapClientContextLock := TCriticalSection.Create;
  FMapClientContextLock.SetLockName('MapClientContextLock');
end;

destructor TIMapClientContext.Destroy;
begin
  FreeClientContexts;
  FreeAndNil(FMapClientContextLock);
  inherited Destroy;
end;

function TIMapClientContext.FindClientContext(Index: Cardinal): TIClientContext;
begin
  Result := nil;
  if ((Index < 1 ) or (Index > FClientContextCount)) then Exit;
  if (FBuckets[Index].IsUsed) then
    Result := FBuckets[Index].ClientContext;
end;

procedure TIMapClientContext.FreeClientContexts;
var
  I: Integer;
  FClientContext: TIClientContext;
begin
  FMapClientContextLock.Lock;
  if FClientContextCount > 0 then
  for i := 1 to FClientContextCount do
  begin
    if not FBuckets[i].IsRelease then
    begin
      FBuckets[i].IsRelease := True;
      FClientContext := FBuckets[i].ClientContext;
      if FClientContext<>nil then FreeAndNil(FClientContext);
    end;
  end;
  FClientContextCount := 0;
  FMapClientContextLock.UnLock;
end;

procedure TIMapClientContext.InitHasTableLength(Count: WORD);
var
  i: Integer;
begin
  if not Assigned(FOnCreateClientEvent) then
    raise Exception.Create('AllocateFreeClientContextFromPool Error, the FOnCreateClientEvent is NULL. ');
  FClientContextCount := Count;
  if (Count = 0) then FClientContextCount := 1000;
  SetLength(FBuckets, FClientContextCount + 1);
  for i := 1 to FClientContextCount do
  begin
    FBuckets[i].ClientContext := FOnCreateClientEvent;
    FBuckets[i].ClientContext.MapClientContext := Self;
    FBuckets[i].IsUsed := False;
    FBuckets[i].IsRelease := False;
    FBuckets[i].MapID := i;
  end;
end;

function TIMapClientContext.RemoveClientContext(
  FClientContext: TIClientContext): Boolean;
begin
  Result := False;
  if ((FClientContext.FMapID < 1 ) or (FClientContext.FMapID > FClientContextCount)) then Exit;
  FMapClientContextLock.Lock;
  with FBuckets[FClientContext.FMapID] do
  begin
    if FBuckets[FClientContext.FMapID].IsUsed then
    begin
      FBuckets[FClientContext.FMapID].IsUsed := False;
      FClientContext.InitClientContext;
      Dec(FActiveCount);
      Result := True;
    end;
  end;
  FMapClientContextLock.UnLock;
end;

end.
